package org.activequant.data.util;

import java.util.Comparator;
import java.util.Iterator;
import java.util.concurrent.atomic.AtomicReference;

import org.activequant.core.domainmodel.marketscanner.MarketScan;
import org.activequant.core.domainmodel.marketscanner.MarketScanSeriesSpecification;
import org.activequant.core.domainmodel.marketscanner.ScanCriteria;
import org.activequant.core.domainmodel.marketscanner.ScanDataEntity;
import org.activequant.core.types.TimeFrame;
import org.activequant.core.types.TimeStamp;
import org.activequant.data.retrieval.IMarketScanSubscription;
import org.activequant.data.retrieval.IMarketScanSubscriptionSource;
import org.activequant.data.retrieval.IScanDataReplayService;
import org.activequant.data.retrieval.IScanSeriesDataIteratorSource;
import org.activequant.math.algorithms.MergeSortIterator;
import org.apache.log4j.Logger;

/**
 * Aligns series of different marketscan entities, and emits them as a single stream of
 * mixed entities: MarketScan for different instruments. 
 * The output sequence presents the order in which events occurred
 * (i.e. different data sources are correctly aligned).
 * <p>
 * In other words, the output is the timeline-based output. See {@link IScanDataReplayService} 
 * for more details. <br/>
 * An IScanDataReplayService implementation. Holds the following associated variables:
 * <ul>
 * <li>currentTimeStamp(AtomicReference&lt;TimeStamp&gt;)</li>
 * <li>comparator(Comparator&lt;ScanDataEntity&lt;?&gt;&gt;)</li>
 * <li>mergeSort(MergeSortIterator&lt;ScanDataEntity&lt;?&gt;&gt;)</li>
 * <li>marketScanReplay(ScanReplayServiceBase&lt;MarketScan&gt;)</li>
 * </ul>
 *  @author Dimitar Misev
 */
public class ScanDataReplayService implements IScanDataReplayService {
	
	private final Logger log = Logger.getLogger(getClass());
	/**
	 * private final AtomicReference&lt;TimeStamp&gt; currentTimeStamp = new AtomicReference&lt;TimeStamp&gt;();
	 */
	private final AtomicReference<TimeStamp> currentTimeStamp = new AtomicReference<TimeStamp>();
	/**
	 * final Comparator&lt;ScanDataEntity&lt;?&gt;&gt; comparator = new Comparator&lt;ScanDataEntity&lt;?&gt;&gt;() {...}<br/>
	 * compares ScanDataEntity subclasses by their timeStamps.
	 */
	final Comparator<ScanDataEntity<?>> comparator = new Comparator<ScanDataEntity<?>>() {
		public int compare(ScanDataEntity<?> o1, ScanDataEntity<?> o2) {
			int rc = o1.getTimeStamp().compareTo(o2.getTimeStamp());
			if(rc != 0) return rc;
			
			// in case market data entities have the same timestamp, make sure that
			// Candles will come last. A candle may have same timestamp as TradeIndication that
			// was generated by this candle (this implies that candle series is used in backtesting
			// context as source of generated trade indications <em>and</em> as data feed).
			// Since in non-simulated context trade indications that are used to build a Candle
			// are always happen <em>before</em> the candle is fired, we better make simulated
			// data feed behave the same way.
//			if(o1.getClass() == Candle.class || o2.getClass() == Candle.class) {
//				if(o1.getClass() == Candle.class) {
//					if(o2.getClass() == Candle.class) {
//						return 0; // candle-to-candle
//					} else {
//						return 1;
//					}
//				} else {
//					return -1;
//				}
//			}
			
			return 0;
		}
	};
	/**
	 * final MergeSortIterator&lt;ScanDataEntity&lt;?&gt;&gt; mergeSort = new MergeSortIterator&lt;ScanDataEntity&lt;?&gt;&gt;(comparator);<br/>
	 * a MergeSortIterator which sorts and merges data from feeds by their timeStamp
	 */
	final MergeSortIterator<ScanDataEntity<?>> mergeSort = new MergeSortIterator<ScanDataEntity<?>>(comparator);
	/**
	 * private final ScanReplayServiceBase&lt;MarketScan&gt; marketScanReplay = new ScanReplayServiceBase&lt;MarketScan&gt;() {...}
	 */
	private final ScanReplayServiceBase<MarketScan> marketScanReplay = new ScanReplayServiceBase<MarketScan>() {
		/**
		 * Creates and returns a MarketScanSeriesSpecification using the instrumentSpecification(InstrumentSpecification) of the given entity(MarketScan) and TIMEFRAME_1_TICK to set its 
		 * associated instrumentSpecification(InstrumentSpecification) and timeFrame(TimeFrame)
		 */
		@Override
		protected MarketScanSeriesSpecification inferSeriesSpecification(MarketScan entity) {
			return new MarketScanSeriesSpecification(entity.getScanCriteria(), TimeFrame.TIMEFRAME_1_TICK);
		}
		/**
		 * <strong>1.</strong> Gets a prot(Iterator&lt;Quote&gt;) to iterate the given iterator(Iterator&lt;MarketScan&gt;) in the interval starting after the value of the associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;)
		 * and ending with the endTimeStamp(TimeStamp) of the associated quoteReplay(ScanReplayServiceBase&lt;MarketScan&gt;)<br/>
		 * <strong>2.</strong> Adds a new Stream object to the streams(Queue&lt;Stream&gt;) of the associated mergeSort(MergeSortIterator&lt;MarketDataEntity&lt;?&gt;&gt;)
		 * using that prot(Iterator&lt;Quote&gt;)
		 */
		@SuppressWarnings("unchecked")
		@Override
		protected void processFetchedIterator(Iterator<MarketScan> iterator) {
			TimeStamp low = new TimeStamp(currentTimeStamp.get().getNanoseconds() + 1L);
			Iterator<MarketScan> prot = new ScanProtectTimeIntervalIterator<MarketScan>(iterator, low, marketScanReplay.getEndTimeStamp());
			mergeSort.addIterator((Iterator<ScanDataEntity<?>>) (Iterator<?>) prot);
		}
	};
	
	/**
	 * Start of the historical interval.<br/>
	 * returns the startTimeStamp(TimeStamp) of the associated marketScanReplay(ReplayServiceBase&lt;Quote&gt;)
	 * @return start.
	 */
	public TimeStamp getStartTimeStamp() {
		return marketScanReplay.getStartTimeStamp();
	}
	/**
	 * <strong>1.</strong> Sets the startTimeStamp(TimeStamp) and value of the currentTimeStamp(AtomicReference&lt;TimeStamp&gt;) of the associated marketScanReplay(ScanReplayServiceBase&lt;Quote&gt;) with the given val(TimeStamp)<br/>
	 * <strong>5.</strong> Sets the value of the associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;) with the given val(TimeStamp)
	 * @param val
	 */
	public void setStartTimeStamp(TimeStamp val) {
		marketScanReplay.setStartTimeStamp(val);
		currentTimeStamp.set(val);
	}
	
	/**
	 * End of the historical interval.<br/>
	 * returns the endTimeStamp(TimeStamp) of the associated marketScanReplay(ScanReplayServiceBase&lt;MarketScan&gt;)
	 * @return end.
	 */
	public TimeStamp getEndTimeStamp() {
		return marketScanReplay.getEndTimeStamp();
	}
	/**
	 * <strong>1.</strong> Sets the endTimeStamp(TimeStamp) of the associated marketScanReplay(ScanReplayServiceBase&lt;MarketScan&gt;) with the given val(TimeStamp)<br/>
	 * @param val
	 */
	public void setEndTimeStamp(TimeStamp val) {
		marketScanReplay.setEndTimeStamp(val);
	}
	/**
	 * Historic data source for quotes.<br/>
	 * returns the seriesSource(ISeriesDataIteratorSource&lt;MarketScan&gt;) of the associated marketScanReplay(ScanReplayServiceBase&lt;MarketScan&gt;)
	 * @return data source.
	 */
	public IScanSeriesDataIteratorSource<MarketScan> getMarketScanSeriesDataSource() {
		return marketScanReplay.getSeriesDataSource();
	}
	
	/**
	 * Sets historic data source for quotes.<br/>
	 * sets the seriesSource(IScanSeriesDataIteratorSource&lt;MarketScan&gt;) of the associated marketScanReplay(ScanReplayServiceBase&lt;MarketScan&gt;) with the given val(ISeriesDataIteratorSource&lt;MarketScan&gt;)
	 * @param val data source.
	 */
	public void setMarketScanSeriesDataSource(IScanSeriesDataIteratorSource<MarketScan> val) {
		marketScanReplay.setSeriesDataSource(val);
	}
	/**
	 * <strong>1.</strong> Sets the value of the associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;) with the startTimeStamp(TimeStamp)
	 * of the associated marketScanReplay(ScanReplayServiceBase&lt;MarketScan&gt;)<br/>
	 * <strong>2.</strong> Creates and returns an Iterator&lt;ScanDataEntity&lt;?&gt;&gt; which gets his next element from the associated mergeSort(MergeSortIterator&lt;ScanDataEntity&lt;?&gt;&gt;)
	 * and sets the value of the associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;) with the timeStamp of that next MarketDataEntity element
	 */
	public Iterator<ScanDataEntity<?>> iterator() {
		currentTimeStamp.set(marketScanReplay.getStartTimeStamp());
		
		return new Iterator<ScanDataEntity<?>>() {
			
			public boolean hasNext() {
				return mergeSort.hasNext();
			}

			public ScanDataEntity<?> next() {
				ScanDataEntity<?> entity = mergeSort.next();
				currentTimeStamp.set(entity.getTimeStamp());
				return entity;
			}

			public void remove() {
				mergeSort.remove();
			}
		};
	}
	/**
	 * returns a new IMarketScanSubscriptionSource which uses the associated quoteReplay(ReplayServiceBase&lt;MarketScan&gt;)
	 */
	public IMarketScanSubscriptionSource getMarketScanSubscriptionSource() {
		return new IMarketScanSubscriptionSource() {
			/**
			 * returns an IMarketScanSubscription&lt;MarketScan&gt;[] array from the associated quoteReplay(ScanReplayServiceBase&lt;MarketScan&gt;)
			 */
			public IMarketScanSubscription<MarketScan>[] getSubscriptions() {
				return marketScanReplay.getSubscriptions();
			}
			/**
			 * creates and returns an IMarketScanSubscription&lt;MarketScan&gt; for the given spec(ScanCriteria) and TIMEFRAME_1_TICK
			 * using the associated quoteReplay(ScanReplayServiceBase&lt;MarketScan&gt;)
			 */
			public IMarketScanSubscription<MarketScan> subscribe(ScanCriteria spec)
					throws Exception {
				return marketScanReplay.subscribe(spec, TimeFrame.TIMEFRAME_1_TICK);
			}
			
			public void reconnect() {
				// TODO Auto-generated method stub
			}
		};
	}
	/**
	 * Depending on whether the given e(ScanDataEntity&lt;?&gt;) is a MarketScan class/subclass it invokes
	 * the <code>marketScanReplay.fire(e)</code> method on one of the associated
	 * marketScanReplay(ScanReplayServiceBase&lt;MarketScan&gt;)
	 */
	public void fire(ScanDataEntity<?> e) {
		log.debug("[dispatcher] fire: " + e);
				
		if(MarketScan.class.isAssignableFrom(e.getClass())) {
			marketScanReplay.fire((MarketScan) (Object) e);
		}
	}
	/**
	 * returns the value of the associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;)
	 */
	public TimeStamp currentTimeStamp() {
		return currentTimeStamp.get();
	}
	/**
	 * returns the size of the streams(Queue&lt;Stream&gt;) of the associated mergeSort(MergeSortIterator&lt;ScanDataEntity&lt;?&gt;&gt;)
	 * @return
	 */
	public int getStreamsSize(){
		return mergeSort.getStreamsSize();
	}
}